package com.codejiwei.flink.zk;

import org.apache.flink.shaded.curator4.org.apache.curator.framework.CuratorFramework;

import java.util.List;
import java.util.Properties;

/**
 * author: codejiwei
 * date: 2023/9/7
 * desc:
 **/
public class ZookeeperTest {
    public static void main(String[] args) throws Exception {

        Properties properties = new Properties();
        properties.setProperty("address", "localhost:32181");
        properties.setProperty("sessionTimeout", "60000");
        properties.setProperty("connectTimeout", "60000");
        properties.setProperty("prefix", "test_metadata");

        CuratorFramework zkClient = ZookeeperUtil.getZkClient(properties);
        System.out.println(zkClient.getNamespace());
        zkClient.start();

//        //TODO get children
        List<String> list = zkClient.getChildren().forPath("/sensor");
        System.out.println(list);
//        //TODO get data
//        String data = new String(zkClient.getData().forPath("/sensor/t1"), StandardCharsets.UTF_8);
//        System.out.println("========================" + data);
//        //TODO set data
//        String meta = "{\"id\": \"sensor1\", \"type\": \"string\"}";
//        Reader reader = new InputStreamReader(Files.newInputStream(Paths.get("src/main/resources/data/transform.json")), StandardCharsets.UTF_8);
//        int ch = 0;
//        StringBuffer sb = new StringBuffer();
//        while ((ch = reader.read()) != -1) {
//            sb.append((char) ch);
//        }
//        meta = sb.toString();
//
//        zkClient.setData().forPath("/sensor/t1", meta.getBytes(StandardCharsets.UTF_8));
//
//        String data1 = new String(zkClient.getData().forPath("/sensor/t1"), StandardCharsets.UTF_8);
//        System.out.println("=======================" + data1);


//        NodeCache nodeCache = new NodeCache(zkClient, "/sensor/t1");
//        nodeCache.getListenable().addListener(new NodeCacheListener() {
//            @Override
//            public void nodeChanged() throws Exception {
//                byte[] data2 = nodeCache.getCurrentData().getData();
//                String data3 = new String(data2, StandardCharsets.UTF_8);
//                System.out.println(data3);
//            }
//        });
//        nodeCache.start();

    }
}
